package cn.com.cennavi.sparkStreamService;

import org.apache.spark.api.java.function.Function2;

import cn.com.cennavi.bean.IndexInputParmBean;

/**
 * 指数计算平均速度的reducebykeyandwindow方法
 * 
 * @Description
 * @author Thinkpad
 * @date 2017年3月9日 下午2:49:37
 */
public class IndexComputerAverageSpeedReduceBykeyAndWindowService implements Function2<IndexInputParmBean, IndexInputParmBean, IndexInputParmBean>{

	
	/** @Fields serialVersionUID: */
	  	
	private static final long serialVersionUID = 1L;

	public IndexInputParmBean call(IndexInputParmBean arg0, IndexInputParmBean arg1) throws Exception {
		// TODO Auto-generated method stub
		if (arg0 == null && arg1 == null) {
			System.out.println("error! null data!");
		} else {
			if (arg0 == null || arg1 == null) {
				System.out.println("one of the two is null!");
			}
		}

		int totalNumber = arg0.linkSimpleNum + arg1.linkSimpleNum;
		double sumSpeed = arg0.speed * arg0.linkSimpleNum + arg1.speed * arg1.linkSimpleNum;
		double aveSpeed = sumSpeed / (double) totalNumber;
		IndexInputParmBean data = new IndexInputParmBean();
		data.linkId = arg0.linkId;
		data.linkSimpleNum = totalNumber;
		data.speed = aveSpeed;
		data.timestamp = arg0.timestamp;
		System.out.println("niuxinzan:" + data.toString());
		return data;
	}

}
